package com.zho.common.consumer;

import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

import com.zho.util.Logit;

public class Consumer<T> implements INotifier, IConsumerController {
    private final Logit log = Logit.getInstance(Consumer.class);
    private final static int MAX_COUNT_DATA_NULL = 10;

    protected volatile String name = "";
    protected volatile boolean shutdown = true;
    protected volatile boolean idel = false;
    protected volatile long waitingMillis = -1L;
    protected volatile int countDataNull = 0;
    protected volatile boolean isWatting = false;

    protected ReentrantLock LOCK = new ReentrantLock(false);
    protected Condition WAITING_ON_EMPTY_DATA = LOCK.newCondition();

    private IInDataService<T> inDataService;
    private IOutDataService<T> outDataService;

    private Thread _currThread;

    public Consumer() {
    }

    /**
     * 
     * @param _inDataService
     * @param _outDataService
     */
    public Consumer(IInDataService<T> _inDataService, IOutDataService<T> _outDataService) {
        this.inDataService = _inDataService;
        this.outDataService = _outDataService;

        if (this.inDataService != null) {
            this.inDataService.setConsumerNotifier(this);
        }
    }

    /**
     * if _waitingMillis > 50, the thread will wait/wake-up in _waitingMillis
     * 
     * @param _inDataService
     * @param _outDataService
     * @param _waitingMillis
     */
    public Consumer(IInDataService<T> _inDataService, IOutDataService<T> _outDataService, long _waitingMillis) {
        this.inDataService = _inDataService;
        this.outDataService = _outDataService;
        this.waitingMillis = _waitingMillis;

        if (this.inDataService != null) {
            this.inDataService.setConsumerNotifier(this);
        }
    }

    /*
     * (non-Javadoc)
     * 
     * @see concurr.INotifier#setIdel()
     */
    public void setIdel() {
        if (log.ison())
            log.debug(":: setIdel() ...");
        idel = true;
    }

    /*
     * (non-Javadoc)
     * 
     * @see concurr.INotifier#sendNotify()
     */
    public boolean sendNotify() {
        final String LOCATION = "Consumer: " + name + " - sendNotify()";
        if (log.ison())
            log.debug(LOCATION);
        boolean rs = true;
        if(isWatting && !shutdown) {
            rs = false;
            try {
                // +++ reset Idel
                idel = false;
    
                // +++ reset countDataNull
                countDataNull = 1;
    
                LOCK.lock();
                WAITING_ON_EMPTY_DATA.signalAll();
                // WAITING_ON_EMPTY_DATA.signal();
//                System.out.println("--- " + name + " --> WAITING_ON_EMPTY_DATA.signalAll()...");
                rs = true;
            } catch (Exception e) {
                rs = false;
                log.error(LOCATION, e);
            } finally {
                LOCK.unlock();
            }
        }
        return rs;
    }

    /*
     * (non-Javadoc)
     * 
     * @see concurr.IConsumerManager#start()
     */
    public boolean start() {
        final String LOCATION = "Consumer: " + name + " - start()";
        if (log.ison())
            log.debug(LOCATION + ":: BEGIN - shutdown:" + shutdown + ", idel:" + idel);
        boolean rs = false;
        // if(shutdown && (_currThread == null || _currThread.isInterrupted()))
        // {
        if (shutdown && _currThread == null) {
            if (log.ison())
                log.debug(LOCATION + ":: ----- new ConsumerTask ..... ");
            _currThread = new Thread(new ConsumerTask(), name);
            _currThread.start();
            rs = true;
        } else {
            if (log.ison())
                log.debug(LOCATION + ":: Started....");
        }
        if (log.ison())
            log.debug(LOCATION + ":: END");
        return rs;
    }

    /*
     * (non-Javadoc)
     * 
     * @see concurr.IConsumerManager#stop()
     */
    public boolean stop() {
        final String LOCATION = "Consumer: " + name + " - stop()";
        if (log.ison())
            log.debug(LOCATION + ":: BEGIN");
        boolean rs = false;
        try {
            shutdown = true;
            countDataNull = 0;
            if (_currThread != null && !_currThread.isInterrupted()) {
                _currThread.interrupt();
                _currThread = null;
                rs = true;
            } else {
                if (log.ison())
                    log.debug(LOCATION + ":: Stopped....");
            }
            if (log.ison())
                log.debug(LOCATION + ":: END");
        } catch (Exception e) {
            if (log.ison())
                log.debug(LOCATION, e);
        }
        return rs;
    }

    /**
     * 
     * @author tungnq.nguyen
     * 
     */
    private class ConsumerTask implements Runnable {
        private final Logit log = Logit.getInstance(ConsumerTask.class);

        /*
         * (non-Javadoc)
         * 
         * @see java.lang.Runnable#run()
         */
        public void run() {
            final String LOCATION = "Consumer: " + name + " - run()";
            log.debug(LOCATION + ":: BEGIN");
            shutdown = false;
            countDataNull = 0;
            // idel = false;
            main_while_loop: while (!shutdown && !Thread.currentThread().isInterrupted()) {
                isWatting = false;
                countDataNull = 0;
                try {
                    if (log.ison())
                        log.debug("Consumer: " + name + " ---  do lock ... countDataNul:" + countDataNull);
                    LOCK.lock();
                    // processing data when in/out data is available...
                    List<T> datas = null;

                    if (log.ison())
                        log.debug("Consumer: " + name + ", shutdown: " + shutdown + ", idel: " + idel + ", inDataService:"
                                        + inDataService.checkAvailable() + ", outDataService:" + outDataService.checkAvailable());
                    
                    while (!shutdown && !idel && countDataNull <= MAX_COUNT_DATA_NULL && inDataService.checkAvailable()
                            && outDataService.checkAvailable()) {
                        // log.debug("--- do business ...");
                        try {                            
                            datas = inDataService.recieve();
                            
                            outDataService.send(datas);
                        } catch (Exception e) {
                            log.warn("run()", e);
                            inDataService.callbackError(datas, e);
                        } finally {
                            if (datas != null && datas.size() > 0) {
                                countDataNull = 1;
                            } else {
                                countDataNull++;
                            }
                        }
                    }

                    // check to see if the input_data is empty - wait if it is.
                    if(!shutdown) {
                        if (waitingMillis > 5) {
                            if (idel) {
                                if (log.ison())
                                    log.debug("Consumer: " + name + "---  Idel: going to await ...");
                                isWatting = true;
                                WAITING_ON_EMPTY_DATA.await();
//                                System.out.println("--- " + name + " --> WAITING_ON_EMPTY_DATA.await()...");
                            } else {
                                if (log.ison())
                                    log.debug("Consumer: " + name + "---  waitingMillis(" + waitingMillis + "): going to await ...");
                                isWatting = true;
                                WAITING_ON_EMPTY_DATA.await(waitingMillis, TimeUnit.MILLISECONDS);
//                                System.out.println("--- " + name + " --> WAITING_ON_EMPTY_DATA.await(waitingMillis)...");
                            }
                        } else {
                            if (log.ison())
                                log.debug("Consumer: " + name + "---  going to await ...");
                            isWatting = true;                            
                            WAITING_ON_EMPTY_DATA.await();
//                            System.out.println("--- " + name + " --> WAITING_ON_EMPTY_DATA.await()...");
                        }
                    }
                    /*
                    while (!shutdown
                            && (countDataNull == 0 
                                    || !(countDataNull < MAX_COUNT_DATA_NULL) 
                                    || !inDataService.checkAvailable() 
                                    || !outDataService.checkAvailable())) {
                        if (waitingMillis > 50) {
                            if (idel) {
                                if (log.ison())
                                    log.debug("Consumer: " + name + "---  Idel: going to await ...");
                                isWatting = true;
                                WAITING_ON_EMPTY_DATA.await();
                                System.out.println("--- " + name + " --> WAITING_ON_EMPTY_DATA.await()...");
                            } else {
                                if (log.ison())
                                    log.debug("Consumer: " + name + "---  waitingMillis(" + waitingMillis + "): going to await ...");
                                isWatting = true;
                                WAITING_ON_EMPTY_DATA.await(waitingMillis, TimeUnit.MILLISECONDS);
                                System.out.println("--- " + name + " --> WAITING_ON_EMPTY_DATA.await(waitingMillis)...");
                            }
                        } else {
                            if (log.ison())
                                log.debug("Consumer: " + name + "---  going to await ...");
                            isWatting = true;                            
                            WAITING_ON_EMPTY_DATA.await();
                            System.out.println("--- " + name + " --> WAITING_ON_EMPTY_DATA.await()...");
                        }
                    }      
                    */              
                } catch (Exception e) {                    
                    log.warn(":: " + Thread.currentThread().getName() + " was interrupted....");
                    // continue main_while_loop;
                    Thread.currentThread().interrupt();
                    // restores the interrupt flag of this thread
                    // (in case shutdown is not set, this will cause the main
                    // loop to stop)
                    break main_while_loop;
                } finally {
                    if (log.ison())
                        log.debug("Consumer: " + name + "---  do unlock ...");
                    LOCK.unlock();
                }
            }
            shutdown = true;
            countDataNull = 0;
            // idel = true;
            log.debug(LOCATION + ":: END");
        }
    }

    /**
     * @param waitingMillis
     *            the waitingMillis to set
     */
    public void setWaitingMillis(long waitingMillis) {
        this.waitingMillis = waitingMillis;
    }

    /**
     * @param inDataService
     *            the inDataService to set
     */
    public void setInDataService(IInDataService<T> inDataService) {
        this.inDataService = inDataService;
        if (this.inDataService != null) {
            this.inDataService.setConsumerNotifier(this);
        }
    }

    /**
     * @param outDataService
     *            the outDataService to set
     */
    public void setOutDataService(IOutDataService<T> outDataService) {
        this.outDataService = outDataService;
    }

    /**
     * @param idel
     *            the idel to set
     */
    public void setIdel(boolean idel) {
        this.idel = idel;
    }

    /**
     * @param name
     *            the name to set
     */
    public void setName(String name) {
        this.name = (name == null ? "" : name);
    }
    
    @Override
    public String getName() {
        return name;
    }

    @Override
    public boolean isIdel() {
        return idel;
    }

    @Override
    public boolean isRunning() {
        return !shutdown;
    }

    @Override
    public boolean isWaiting() {
        return isWatting;
    }
}
